/**
 * @(#)ThreadPoolExecutor.java 2017年12月26日
 * Copyright(C) 2017 ZHEXIN IT CO.,LTD. All rights reserved.
 */
package org.jsbd.boss.util.threads;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.tomcat.util.threads.TaskQueue;
import org.jsbd.boss.common.redis.RedisClient;
import org.jsbd.boss.domian.channel.MiguUser;
import org.jsbd.boss.task.MiguUserCheckThread;
import org.jsbd.boss.util.CacheKey;
import org.jsbd.boss.util.SpringContextUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * TODO 请加入本类的说明
 * 
 * @author <a href="mailto:xupengcheng@zhexinit.com" >徐鹏程</a>
 * @version 1.0.0
 */
public class ThreadPoolExecutor extends java.util.concurrent.ThreadPoolExecutor {
	public static final Logger LOGGER = LoggerFactory.getLogger(ThreadPoolExecutor.class);
	
	/**
	 * The number of tasks submitted but not yet finished. This includes tasks
	 * in the queue and tasks that have been handed to a worker thread but the
	 * latter did not start executing the task yet. This number is always
	 * greater or equal to {@link #getActiveCount()}.
	 */
	private final AtomicInteger submittedCount = new AtomicInteger(0);

	public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {
		super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
	}

	public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) {
		super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
	}

	public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {
		super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, new RejectHandler());
	}

	public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
		super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, new RejectHandler());
	}

	@Override
	protected void afterExecute(Runnable r, Throwable t) {
		submittedCount.decrementAndGet();
	}

	public int getSubmittedCount() {
		return submittedCount.get();
	}

	/**
	 * {@inheritDoc}
	 */
	@Override
	public void execute(Runnable command) {
		execute(command, 0, TimeUnit.MILLISECONDS);
	}

	/**
	 * Executes the given command at some time in the future. The command may
	 * execute in a new thread, in a pooled thread, or in the calling thread, at
	 * the discretion of the <tt>Executor</tt> implementation. If no threads are
	 * available, it will be added to the work queue. If the work queue is full,
	 * the system will wait for the specified time and it throw a
	 * RejectedExecutionException if the queue is still full after that.
	 *
	 * @param command
	 *            the runnable task
	 * @throws RejectedExecutionException
	 *             if this task cannot be accepted for execution - the queue is
	 *             full
	 * @throws NullPointerException
	 *             if command or unit is null
	 */
	public void execute(Runnable command, long timeout, TimeUnit unit) {
		submittedCount.incrementAndGet();
		try {
			super.execute(command);
		} catch (RejectedExecutionException rx) {
			if (super.getQueue() instanceof TaskQueue) {
				final TaskQueue queue = (TaskQueue) super.getQueue();
				try {
					if (!queue.force(command, timeout, unit)) {
						submittedCount.decrementAndGet();
						throw new RejectedExecutionException("Queue capacity is full.");
					}
				} catch (InterruptedException x) {
					submittedCount.decrementAndGet();
					Thread.interrupted();
					throw new RejectedExecutionException(x);
				}
			} else {
				submittedCount.decrementAndGet();
				throw rx;
			}

		}
	}

	private static class RejectHandler implements RejectedExecutionHandler {
		@Override
		public void rejectedExecution(Runnable r, java.util.concurrent.ThreadPoolExecutor executor) {
			if(r instanceof MiguUserCheckThread){
		 	    RedisClient redisClient = SpringContextUtils.getBean("redisClient");
				MiguUserCheckThread checkThread = (MiguUserCheckThread) r;
				MiguUser miguUser = checkThread.getMiguUser();
				miguUser.setSystemAppName(null);
				miguUser.setAppName(null);
				redisClient.putToQueue(CacheKey.MIGU_CHECK_USER_LIST, miguUser);
			}
			throw new RejectedExecutionException();
		}

	}

}
